package com.hylanda.processor;

import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.Statement;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.nifi.annotation.lifecycle.OnRemoved;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.InputStreamCallback;
import org.apache.nifi.stream.io.StreamUtils;

import com.alibaba.fastjson.JSONObject;
import com.hylanda.processor.database.util.Constant;
import com.hylanda.processor.database.util.SqlServerHelper;

public abstract class ReportProcessor extends AbstractProcessor {
	static final String YES = "是";
	static final String NO = "否";
	static final String COLUMN_PREFIX = "values_";
	static final String VALUES = "[values]";
	private final String dataTable;
	private final String cfgTable;
	private final int columnSize;

	public static final PropertyDescriptor CFG_DESCRIPTOR = new PropertyDescriptor.Builder().name("cfg")
			.displayName("模型显示配置").required(true).allowableValues("        ").defaultValue("        ").build();

	public static final PropertyDescriptor DATA_DESCRIPTOR = new PropertyDescriptor.Builder().name("data")
			.displayName("模型数据配置").required(true).allowableValues("        ").defaultValue("        ").build();

	public static final PropertyDescriptor CONTROL_DESCRIPTOR = new PropertyDescriptor.Builder().name("control")
			.displayName("模型控制").required(true).allowableValues("        ").defaultValue("        ").build();

	public static final PropertyDescriptor CONTROL_VALUES = new PropertyDescriptor.Builder()
			.name("control-values").displayName("◆            启动时是否清空已有模型数据").description("启动时默认不清空已有模型数据")
			.required(false).allowableValues(YES, NO).defaultValue(NO).build();

	static final Relationship REL_SUCCESS = new Relationship.Builder().name("成功")
			.description("成功的更新了模型显示配置等，输入数据成功保存").build();

	static final Relationship REL_FAILURE = new Relationship.Builder().name("失败")
			.description("未成功更新模型显示配置，输入数据未成功保存，输入数据非法").build();
	protected List<PropertyDescriptor> descriptors;
	private Set<Relationship> relationships;

	public ReportProcessor(String dataTable, String cfgTable, int columnSize) {
		super();
		this.dataTable = dataTable;
		this.cfgTable = cfgTable;
		this.columnSize = columnSize;
	}

	@Override
	protected void init(final ProcessorInitializationContext context) {
		final Set<Relationship> relationships = new HashSet<Relationship>();
		relationships.add(REL_SUCCESS);
		relationships.add(REL_FAILURE);
		this.relationships = Collections.unmodifiableSet(relationships);
	}

	@Override
	public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
		return descriptors;
	}

	@Override
	public Set<Relationship> getRelationships() {
		return this.relationships;
	}

	@OnScheduled
	public void onScheduled(final ProcessContext context) {
		// 是否清空已有数据
		final boolean empty = context.getProperty(CONTROL_VALUES).getValue().equals(YES);
		try (final Connection conn = (Connection) SqlServerHelper.connectDb(Constant.hostname,
				Constant.databaseName, Constant.username, Constant.password);
				final Statement stmt = SqlServerHelper.getStatement(conn)) {
			// 清空旧数据
			if (empty) {
				stmt.executeUpdate("update " + dataTable + " set b_flag=0 where processor_id='" + getIdentifier()
						+ "'");
			}
		} catch (Exception e) {
			getLogger().info("清空旧数据失败，processor_id=" + getIdentifier());
			e.printStackTrace();
		}
	}

	@OnRemoved
	public void onRemoved() {
		try (final Connection conn = (Connection) SqlServerHelper.connectDb(Constant.hostname,
				Constant.databaseName, Constant.username, Constant.password);
				final Statement stmt = SqlServerHelper.getStatement(conn)) {
			conn.setAutoCommit(false);
			stmt.addBatch("delete from " + cfgTable + " where processor_id='" + getIdentifier() + "'");
			stmt.addBatch("update " + dataTable + " set b_flag=0 where processor_id='" + getIdentifier() + "'");
			stmt.executeBatch();
			conn.commit();
		} catch (Exception e) {
			getLogger().info("删除模型数据失败，processor_id=" + getIdentifier());
			e.printStackTrace();
		}
	}

	protected JSONObject readFromInput(final ProcessSession session, FlowFile flowFile) throws Exception {
		// 读取flowFile内容
		final byte[] buffer = new byte[(int) flowFile.getSize()];
		session.read(flowFile, new InputStreamCallback() {
			@Override
			public void process(final InputStream in) throws IOException {
				StreamUtils.fillBuffer(in, buffer);
			}
		});

		// 解析输入为json数组
		final JSONObject obj;
		try {
			obj = JSONObject.parseObject(new String(buffer, StandardCharsets.UTF_8));
		} catch (Exception e) {
			throw new Exception("input flowfile is not json");
		}
		return obj;
	}

	protected void saveToDB(final StringBuilder sqlAccumulatorCfg, final StringBuilder sqlAccumulator)
			throws Exception {
		try (final Connection con = (Connection) SqlServerHelper.connectDb(Constant.hostname,
				Constant.databaseName, Constant.username, Constant.password);
				final Statement stmt = SqlServerHelper.getStatement(con)) {
			stmt.executeUpdate(sqlAccumulatorCfg.toString());
			sqlAccumulatorCfg.setLength(0);
			if (sqlAccumulator.length() > 0) {
				stmt.executeUpdate(sqlAccumulator.toString());
				sqlAccumulator.setLength(0);
			}
		} catch (Exception e) {
			throw e;
		}
	}

	@Override
	public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
		FlowFile flowFile = session.get();
		if (flowFile == null) {
			return;
		}
		final ComponentLog logger = getLogger();

		// 解析数据信息
		final String[] columnAnalyzed = new String[columnSize];
		for (int i = 0; i < columnSize; i++) {
			columnAnalyzed[i] = context.getProperty(COLUMN_PREFIX + i).getValue();
		}

		// 解析输入为json数组
		final JSONObject obj;
		try {
			obj = readFromInput(session, flowFile);
		} catch (Exception e) {
			logger.error(e.getMessage());
			session.transfer(flowFile, REL_FAILURE);
			return;
		}

		final String[] columns = new String[columnSize];
		for (int i = 0; i < columnSize; i++) {
			columns[i] = obj.getString(columnAnalyzed[i]);
		}

		// 保存信息
		try {
			this.saveCfg(context, columns);
		} catch (Exception e1) {
			logger.error("saveToDB error {}", new Object[] { e1.getMessage() });
			session.transfer(flowFile, REL_FAILURE);
			return;
		}
		session.transfer(flowFile, REL_SUCCESS);
	}

	abstract protected void saveCfg(ProcessContext context, String[] columns) throws Exception;

}
